package filter;

import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 2 * @Author: 王杰
 * 3 * @Date: 2020/11/18 12:52
 * 4
 */
public class Demo01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> input = env.readTextFile("data/ip31SampleEventof1_Schema.csv");


        SingleOutputStreamOperator<String> result = input.setParallelism(1).filter(new RichFilterFunction<String>() {

            private ValueState<Boolean> fisrtline = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                fisrtline = getRuntimeContext().getState(new ValueStateDescriptor<>("fisrtline", Boolean.class));
            }

            @Override
            public boolean filter(String value) throws Exception {
                if (fisrtline.value() == null) {
                    fisrtline.update(Boolean.TRUE);
                    return true;
                }
                if (value.contains("SampleTime")) {
                    return false;
                }
                return true;
            }
        });

        result.print();
        env.execute();
    }
}
